package com.shop;

import com.mapreduce.JoinReduce;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.net.URI;

public class Step8 extends Configured implements Tool {
    //TextInputFormat->原始数据/shopdata
    //k1			v1
    //0		4,10893,paySuccess,1713755898058
    //...		4,10890,paySuccess,1713755898059
    //		4,10868,paySuccess,1713755898059
    //
    //->
    //k2			v2
    //4,10893		paySuccess
    //4,10890		paySuccess
    //4,10868		paySuccess
    static class Step8OneMapper extends Mapper<LongWritable,Text,Text,Text>{
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
            String[] datas = value.toString().split(",");
            if (datas[2].equals("paySuccess")){
                context.write(new Text(datas[0]+","+datas[1]),new Text(datas[2]));
            }

        }
    }
    //SequenceFileInputFormat->Step7的数据
    //k1		v1
    //4,10869	1.05
    //4,10876	2.0
    //4,10890	2.05
    //4,10893	3.15
    //->
    //k2		v2
    //4,10869	1.05
    //4,10876	2.0
    //4,10890	2.05
    //4,10893	3.15
    //4,10893		paySuccess
    //4,10890		paySuccess
    //4,10868		paySuccess
    static class Step8TwoMapper extends Mapper<Text,DoubleWritable,Text,Text>{
        @Override
        protected void map(Text key, DoubleWritable value, Mapper<Text, DoubleWritable, Text, Text>.Context context) throws IOException, InterruptedException {
            context.write(key,new Text(value.toString()));
        }
    }
    //->
    //4,10893			[3.15,paySuccess]
    //4,10890			[2.05,paySuccess]
    //4,10868			[paySuccess]
    //4,10869			[1.05]
    //->
    //k3				v3
    //4,10869			1.05
    static class Step8Reducer extends Reducer<Text,Text,Text,DoubleWritable>{
        @Override
        protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {
            Text result = new Text();
            for (Text value : values) {
                result= value;
                if (result.toString().equals("paySuccess")){
                    return;
                }
            }
            context.write(key,new DoubleWritable(Double.valueOf(result.toString())));
        }
    }
    @Override
    public int run(String[] strings) throws Exception {
        Configuration conf = getConf();
        // 指定输入输出路径
        Path step8input1 = new Path(
                "hdfs://192.168.10.11:9000/shopdata");
        Path step8input2 = new Path(
                "hdfs://192.168.10.11:9000/shop7output");
        Path step8output = new Path(
                "hdfs://192.168.10.11:9000/shop8output");

        FileSystem fs = FileSystem.get(
                new URI("hdfs://192.168.10.11:9000")
                , conf);
        if (fs.exists(step8output)) fs.delete(step8output, true);
        //构建Job
        Job job = Job.getInstance(conf);
        job.setJobName("step8");
        job.setJarByClass(this.getClass());

        // 设置多个并行的Map
        MultipleInputs.addInputPath(job,step8input1,
                TextInputFormat.class,
                Step8OneMapper.class);
        MultipleInputs.addInputPath(job,step8input2,
                SequenceFileInputFormat.class,
                Step8TwoMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(Step8Reducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);


//        job.setOutputFormatClass(TextOutputFormat.class);
//        TextOutputFormat.setOutputPath(job, step8output);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setOutputPath(job, step8output);
        return job.waitForCompletion(true) ? 0 : -1;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new Step8(),args));
    }
}
